package com.web.basic.service.impl;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.StringReader;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.web.basic.service.ElasticsearchService;
import com.web.common.basic.elasticsearch.entity.ElasticDataListParam;
import com.web.common.basic.elasticsearch.entity.ElasticIndexParam;
import com.web.common.basic.elasticsearch.entity.ElasticIndices;
import com.web.common.basic.elasticsearch.entity.ElasticParam;
import com.web.common.basic.entity.ElasticsearchIndex;
import com.web.common.basic.entity.SimplePage;
import com.web.common.result.R;
import com.web.common.result.ResultEnum;
import com.web.common.tool.DateTimeUtil;
import com.web.common.tool.Symbols;
import com.web.service.handler.error.CustomException;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;

@Service
public class ElasticsearchServiceImpl implements ElasticsearchService {

	@Autowired
	private RestClient elRestClient;

	@Autowired
	private ElasticsearchClient elasticsearchClient;
	
	private static final List<Integer> HTTP_SUCCESS_LIST = new ArrayList<>();
	
	private static final ObjectMapper MAPPER = new ObjectMapper();
	static {
		MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
		
		HTTP_SUCCESS_LIST.add(HttpStatus.OK.value());
		HTTP_SUCCESS_LIST.add(HttpStatus.CREATED.value());
	}
	
	@Override
	public R<Object> customOperate(ElasticParam param) {
		R<Object> json = new R<>();
		StringBuilder stBuilder = getCustomOperateResult(param);
		
		json.setResultEnum(ResultEnum.SUCCESS);
		json.setData(stBuilder.toString());
		return json;
	}
	
	/**
	 * 获取自定义操作的处理结果，转成字符串
	 * @param param 请求操作的信息
	 * @return StringBuilder 结果字符串
	 *
	 * @author zhouhui
	 * @since 1.0.0
	 */
	private StringBuilder getCustomOperateResult(ElasticParam param) {
		Response res = null;
		try {
			Request request = new Request(param.getMethod(), param.getEndpoint());
			
			if(StringUtils.hasText(param.getJsonEntity())) {
				request.setJsonEntity(param.getJsonEntity());
			}
			if(param.getParameters() != null && !param.getParameters().isEmpty()) {
				request.addParameters(param.getParameters());
			}
			res = elRestClient.performRequest(request);
		} catch (Exception e) {
			e.printStackTrace();
		}
		if (res == null) {
			throw new CustomException(ResultEnum.HTTP_NO_RESPONSE.getMessage());
		}
		int statusCode = res.getStatusLine().getStatusCode();
		if (!HTTP_SUCCESS_LIST.contains(statusCode)) {
			throw new CustomException("请求执行失败");
		}
		StringBuilder stBuilder = new StringBuilder();
		if(res.getEntity() != null) {
			try {
				InputStream content = res.getEntity().getContent();
				InputStreamReader reader = new InputStreamReader(content);
				BufferedReader bf = new BufferedReader(reader);
				String str = null;
				while((str = bf.readLine()) != null) {
					stBuilder.append(str);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		return stBuilder;
	}

	@Override
	public R<List<ElasticIndices>> getIndexList() {
		R<List<ElasticIndices>> json = new R<>();
		
		ElasticParam param = new ElasticParam();
		param.setMethod(HttpMethod.GET.name());
		param.setEndpoint("/_cat/indices/"+ ElasticsearchIndex.INDEX_PREFIX +"*");
		Map<String, String> parameters = new HashMap<>();
		parameters.put("format", "json");
		param.setParameters(parameters);
		
		R<Object> result = customOperate(param);
		if(!ResultEnum.SUCCESS.getCode().equals(result.getCode())) {
			json.setCode(result.getCode());
			json.setMessage(result.getMessage());
			return json;
		}
		List<ElasticIndices> dataList = null;
		try {
			dataList = MAPPER.readValue(result.getData().toString(), new TypeReference<List<ElasticIndices>>() {});
		} catch (Exception e) {
			e.printStackTrace();
		}
		json.setResultEnum(ResultEnum.SUCCESS);
		json.setData(dataList);
		return json;
	}
	
	@Override
	public R<Map<String, Object>> getIndex(ElasticIndexParam param) {
		R<Map<String, Object>> json = new R<>();
		
		ElasticParam createParam = new ElasticParam();
		createParam.setMethod(HttpMethod.GET.name());
		createParam.setEndpoint(Symbols.FORWARD_SLASH + param.getIndex());
		
		R<Object> result = customOperate(createParam);
		if(!ResultEnum.SUCCESS.getCode().equals(result.getCode())) {
			json.setCode(result.getCode());
			json.setMessage(result.getMessage());
			return json;
		}
		if(result.getData() == null) {
			throw new CustomException(ResultEnum.HTTP_REQUEST_ERROR.getMessage());
		}
		Map<String, Object> indexConfig = null;
		try {
			indexConfig = MAPPER.readValue(result.getData().toString(), new TypeReference<Map<String, Object>>() {});
		} catch (JsonProcessingException e) {
			e.printStackTrace();
		}
		json.setResultEnum(ResultEnum.SUCCESS);
		json.setData(indexConfig);
		return json;
	}

	@Override
	public R<Object> createIndex(ElasticIndexParam param) {
		R<Object> json = new R<>();
		//限制一下索引的格式
		if(!param.getIndex().startsWith(ElasticsearchIndex.INDEX_PREFIX)) {
			json.setResultEnum(ResultEnum.ERROR);
			json.setMessage("索引需要以" + ElasticsearchIndex.INDEX_PREFIX + "开头");
			return json;
		}
		
		if(indexExist(param.getIndex())) {
			json.setResultEnum(ResultEnum.ERROR);
			json.setMessage("索引已经存在");
			return json;
		}
		ElasticParam createParam = new ElasticParam();
		createParam.setMethod(HttpMethod.PUT.name());
		createParam.setEndpoint(Symbols.FORWARD_SLASH + param.getIndex());
		createParam.setJsonEntity(param.getConfig());
		R<Object> result = customOperate(createParam);
		
		if(!ResultEnum.SUCCESS.getCode().equals(result.getCode())) {
			json.setCode(result.getCode());
			json.setMessage(result.getMessage());
			return json;
		}
		json.setResultEnum(ResultEnum.SUCCESS);
		return json;
	}

	@Override
	public R<Object> periodicDeleteLog() {
		R<Object> json = new R<>();
		//定期删除两年前的数据
		try {
			LocalDate curDate = LocalDate.now();
			String deleteDate = curDate.minusYears(2).format(DateTimeFormatter.ofPattern(DateTimeUtil.YEAR_MONTH_DATE));
			Query query = Query.of(q -> q.match(t -> t.field("createDate").query(deleteDate)));
			elasticsearchClient.deleteByQuery(fn -> fn.index(ElasticsearchIndex.LOG_INDEX).query(query)).deleted();
		} catch (Exception e) {
			e.printStackTrace();
		}
		json.setResultEnum(ResultEnum.SUCCESS);
		return json;
	}
	
	/**
	 * 判断索引是否存在
	 * @param index 索引值
	 * @return boolean true存在;false不存在
	 *
	 * @author zhouhui
	 * @since 1.0.0
	 */
	private boolean indexExist(String index) {
		Request request = new Request(HttpMethod.HEAD.name(), index);
		try {
			Response res = elRestClient.performRequest(request);
			if (res == null) {
				throw new CustomException(ResultEnum.HTTP_NO_RESPONSE.getMessage());
			}
			int statusCode = res.getStatusLine().getStatusCode();
			if(HttpStatus.OK.value() == statusCode) {
				return true;
			}else if(HttpStatus.NOT_FOUND.value() == statusCode) {
				return false;
			}else {
				throw new CustomException("查询索引异常");
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	@Override
	public R<Object> updateIndexMapping(ElasticIndexParam param) {
		R<Object> json = new R<>();
		if(!indexExist(param.getIndex())) {
			json.setResultEnum(ResultEnum.ERROR);
			json.setMessage("索引不存在");
			return json;
		}
		
		ElasticParam createParam = new ElasticParam();
		createParam.setMethod(HttpMethod.PUT.name());
		createParam.setEndpoint(Symbols.FORWARD_SLASH + param.getIndex() + "/_mapping");
		createParam.setJsonEntity(param.getConfig());
		R<Object> result = customOperate(createParam);
		if(!ResultEnum.SUCCESS.getCode().equals(result.getCode())) {
			json.setCode(result.getCode());
			json.setMessage(result.getMessage());
			return json;
		}
		json.setResultEnum(ResultEnum.SUCCESS);
		return json;
	}
	
	@Override
	public R<Integer> getIndexCount(ElasticIndexParam param) {
		R<Integer> json = new R<>();
		ElasticParam createParam = new ElasticParam();
		createParam.setMethod(HttpMethod.GET.name());
		createParam.setEndpoint(Symbols.FORWARD_SLASH + param.getIndex() + "/_count");
		createParam.setJsonEntity(param.getConfig());
		R<Object> result = customOperate(createParam);
		if(!ResultEnum.SUCCESS.getCode().equals(result.getCode())) {
			json.setCode(result.getCode());
			json.setMessage(result.getMessage());
			return json;
		}
		try {
			Map<String, Object> dataMap = MAPPER.readValue(result.getData().toString(), new TypeReference<Map<String, Object>>() {});
			Integer count = Integer.valueOf(dataMap.get("count").toString());
			
			json.setResultEnum(ResultEnum.SUCCESS);
			json.setData(count);
			return json;
		} catch (JsonProcessingException e) {
			e.printStackTrace();
		}
		json.setResultEnum(ResultEnum.JSON_ERROR);
		return json;
	}

	@Override
	public R<Object> deleteIndex(ElasticIndexParam param) {
		R<Object> json = new R<>();
		if(!indexExist(param.getIndex())) {
			json.setResultEnum(ResultEnum.ERROR);
			json.setMessage("索引不存在");
			return json;
		}
		//判断是否保存数据
		R<Integer> countResult = getIndexCount(param);
		if(!ResultEnum.SUCCESS.getCode().equals(countResult.getCode())) {
			json.setResultEnum(ResultEnum.ERROR);
			json.setMessage("获取索引对应的数据量失败");
			return json;
		}
		if(countResult.getData().intValue() > 0) {
			json.setResultEnum(ResultEnum.ERROR);
			json.setMessage("获取索引已经存储数据，不能进行删除");
			return json;
		}
		
		ElasticParam createParam = new ElasticParam();
		createParam.setMethod(HttpMethod.DELETE.name());
		createParam.setEndpoint(Symbols.FORWARD_SLASH + param.getIndex());
		createParam.setJsonEntity(param.getConfig());
		R<Object> result = customOperate(createParam);
		if(!ResultEnum.SUCCESS.getCode().equals(result.getCode())) {
			json.setCode(result.getCode());
			json.setMessage(result.getMessage());
			return json;
		}
		json.setResultEnum(ResultEnum.SUCCESS);
		return json;
	}

	@Override
	public R<SimplePage> getDataList(ElasticDataListParam condition) {
		SearchRequest request = SearchRequest
				.of(fn -> fn.index(condition.getIndex()).withJson(new StringReader(condition.getJsonData())));
		SearchResponse<Object> result = null;
		try {
			result = elasticsearchClient.search(request, Object.class);

		} catch (Exception e) {
			e.printStackTrace();
		}
		if (result == null) {
			throw new CustomException(ResultEnum.DATA_QUERY_ERROR.getMessage());
		}
		List<Hit<Object>> hists = result.hits().hits();
		List<Object> dataList = new ArrayList<>(hists.size());
		for(Hit<Object> item: hists) {
			dataList.add(item.source());
		}
		
		SimplePage page = new SimplePage();
		page.setPage(condition.getPage());
		page.setSize(condition.getSize());
		page.setTotal(result.hits().total().value());
		try {
			page.setRecords(MAPPER.writeValueAsString(dataList));
		} catch (JsonProcessingException e) {
			e.printStackTrace();
		}
		return R.ok(page);
	}
}
